using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using LiteDB;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Timer = System.Timers.Timer;

namespace Cmc.RabbitMQ.Base
{
    public class Connection : IMQConnection
    {
        /// <summary>
        /// 连接
        /// </summary>
        protected IConnection _connection;
        /// <summary>
        /// 频道
        /// </summary>
        protected IModel _channel;

        /// <summary>
        /// 连接选项
        /// </summary>
        public ConnectConfig ConnectConfig { get; }

        public bool IsConnected;

        public string HostName
        {
            get { return ConnectConfig?.HostName; }
        }

        public List<Consumer> Consumers { get; } = new List<Consumer>();

        /// <summary>
        /// 定时器
        /// </summary>
        protected Timer _timer;
        /// <summary>
        /// 发送失败数据库
        /// </summary>
        protected LiteDatabase litedb;

        public Connection(ConnectConfig connectConfig)
        {
            ConnectConfig = connectConfig;
        }
        public bool Connected()
        {

            try
            {
                IsConnected = false;
                var factory = new ConnectionFactory()
                {
                    HostName = ConnectConfig.HostAddress,
                    UserName = ConnectConfig.UserName,
                    Password = ConnectConfig.Password,
                    AutomaticRecoveryEnabled = true
                };
                _connection = factory.CreateConnection();

                _channel = _connection.CreateModel();
                //启动消息发送确认机制，即需要收到RabbitMQ服务端的确认消息
                _channel.ConfirmSelect();
                //_rabbitConnectOptions.Connection = _connection;
                //_rabbitConnectOptions.Channel = _channel; 

                //交换机声明
                foreach (ExchangeConfig exchangeConfig in ConnectConfig.ExchangeConfigs)
                {
                    _channel.ExchangeDeclare(exchangeConfig.ExchangeName, exchangeConfig.ExchangeType.ToString(), exchangeConfig.IsDurable);
                }

                foreach (QueueConfig queueConfig in ConnectConfig.QueueConfigs)
                {
                    _channel.QueueDeclare(queueConfig.QueueName, queueConfig.IsDurable);  //初始化队列声明
                    if (queueConfig.RoutingKey != null)
                    {
                        foreach (string routingKey in queueConfig.RoutingKey)
                        {
                            _channel.QueueBind(queueConfig.QueueName, queueConfig.ExchangeName, routingKey); //初始化绑定路由
                        }
                    }
                }


                foreach (Consumer consumer in Consumers)
                {
                    consumer.ConsumerTag = null;
                    Consume(consumer);

                }

                string path = AppDomain.CurrentDomain.BaseDirectory + "temp.db";
                litedb = new LiteDatabase(path);
                _timer = new Timer(10000);
                _timer.Elapsed += _timer_Elapsed;
                _timer.Start();

                IsConnected = _connection.IsOpen;

            }
            catch (Exception e)
            {

                IsConnected = false;
            }

            return IsConnected;
        }

        private void _timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            if (!IsConnected)   
            {
                Connected();
            } 

            if (IsConnected)
            {
                UpStoreData();
                var con = Consumers.Where(m => string.IsNullOrEmpty(m.ConsumerTag)).ToList();
                foreach (Consumer consumer in con)
                {
                    Consume(consumer);
                }
            }
        }

        public bool PublishByRoutingKey(string routingKey, object message)
        {
            try
            {
                if (!IsConnected)
                {
                    throw new Exception($"MQ服务{ConnectConfig.HostAddress}:{ConnectConfig.Port}未连接！");
                }

                string msgStr = JsonConvert.SerializeObject(message);
                //设置消息持久性
                //IBasicProperties props = _channel.CreateBasicProperties();
                //props.ContentType = "text/plain";
                //props.DeliveryMode = 2;//持久性
                //消息内容转码，并发送至服务器
                var messageBody = System.Text.Encoding.UTF8.GetBytes(msgStr);
                var exchanges = ConnectConfig.QueueConfigs.Where(m => m.RoutingKey != null && m.RoutingKey.Contains(routingKey)).Select(m => m.ExchangeName);
                foreach (string exchange in exchanges)
                {
                    _channel.BasicPublish(exchange, routingKey, null, messageBody);
                }
                //等待确认
                return _channel.WaitForConfirms();
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
                StoreData(routingKey, message);
                return false;
            }
        }

        public bool Publish(string queneName, object message)
        {
            try
            {
                if (!IsConnected)
                {
                    throw new Exception($"MQ服务{ConnectConfig.HostAddress}:{ConnectConfig.Port}未连接！");
                }

                var queue = ConnectConfig.QueueConfigs.FirstOrDefault(m => m.QueueName == queneName);

                if (queue != null)
                {
                    string msgStr = JsonConvert.SerializeObject(message);
                    //设置消息持久性
                    //IBasicProperties props = _channel.CreateBasicProperties();
                    //props.ContentType = "text/plain";
                    //props.DeliveryMode = 2;//持久性
                    _channel.BasicPublish(queue.ExchangeName ?? "", queneName, null, System.Text.Encoding.UTF8.GetBytes(msgStr));
                }
                return _channel.WaitForConfirms();
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
                StoreData(queneName, message);
                return false;
            }
        }
 
        public string Consume(Consumer consumer)
        {
            try
            {


                Consumer con = Consumers.FirstOrDefault(m => m.ConsumerID == consumer.ConsumerID);
                if (con == null)
                {
                    Consumers.Add(consumer);
                }
                else if (!string.IsNullOrWhiteSpace(con.ConsumerTag))
                {
                    return con.ConsumerTag;
                }

                if (!IsConnected)
                {
                    throw new Exception($"MQ服务{ConnectConfig.HostAddress}:{ConnectConfig.Port}未连接！");
                }

                _channel.BasicQos(0, 1, false);

                var consumer1 = new EventingBasicConsumer(_channel);
                //回应消息监控
                consumer1.Received += (sender, e) =>
                {
                    try
                    {
                        string message = Encoding.UTF8.GetString(e.Body.ToArray());
                        Console.WriteLine($"{consumer.QueueName}：接收数据：{message}");
                        // logger.LogInformation($"接收队列：{queueName}：{e.Body}");
                        if (consumer!.ConsumEventHandler != null)
                        {
                            consumer.ConsumEventHandler(sender, e);
                        }
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine($"订阅数据处理错误：" + ex.ToString());
                    }
                    finally
                    {
                        _channel.BasicAck(e.DeliveryTag, false);
                    }
                };
                //绑定消费者
                string consumeTag = _channel.BasicConsume(consumer.QueueName, //队列名
                                                                                            false,    //false：手动应答；true：自动应答
                                                                                                consumer1);
                consumer.ConsumerTag = consumeTag;
                Console.WriteLine($"已经订阅RabbitMQ服务 id:{consumeTag}，队列" + consumer.QueueName);
                return consumeTag;
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
                //if (string.IsNullOrWhiteSpace(consumer.ConsumerTag))
                //{
                //    return Consume(consumer);
                //}
                return consumer.ConsumerTag;
            }
        }



        /// <summary>
        /// 存储数据到本地文件
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="routingKey"></param>
        protected void StoreData(string routingKey, object msg)
        {
            var data = litedb.GetCollection<StoreModel>("StoreModel");
            StoreModel entity = new StoreModel()
            {
                ObjectId = ObjectId.NewObjectId(),
                Storetime = DateTime.Now,
                Msg = msg,
                HostName = HostName,
                RoutingKey = routingKey
            };
            Console.WriteLine("保存数据：" + msg);

            data.Insert(entity);
        }

        /// <summary>
        /// 提交储存数据
        /// </summary>
        protected void UpStoreData()
        {
            try
            {
                var data = litedb?.GetCollection<StoreModel>("StoreModel");
                if (data != null && data.Count() > 0)
                {
                    var list = data.Find(m => m.HostName == HostName).ToList();
                    foreach (var item in list)
                    {
                        try
                        {
                            Console.WriteLine("补发数据:" + item.Msg);
                            PublishByRoutingKey(item.RoutingKey, item.Msg);
                        }
                        finally
                        {
                            data.Delete(item.ObjectId);
                        }
                    }
                }
            }
            catch (Exception ex)
            {

            }
        }
    }
}
